package com.blog.service.impl;

import com.blog.dto.ArticleDTO;
import com.blog.dto.StringMessageDTO;
import com.blog.entity.Article;
import com.blog.entity.Users;
import com.blog.mapper.ArticleMapper;
import com.blog.mapper.BlogTagMapper;
import com.blog.mapper.UsersMapper;
import com.blog.service.AsyncArticleSyncService;
import com.blog.service.KafkaConsumerService;
import com.blog.utils.ArticleSearchUtils;
import com.blog.utils.JwtTokenProvider;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;

import static com.blog.utils.Staticmessage.ADDARTICLEANDES;
import static com.blog.utils.Staticmessage.INITESDATA;

@Service
@Slf4j
public class KafkaConsumerServiceImpl implements KafkaConsumerService {
    @Resource
    private AsyncArticleSyncService asyncArticleSyncService;
    @Resource
    private ObjectMapper objectMapper;
    @Resource
    private BlogTagMapper blogTagMapper;
    @Resource
    private UsersMapper usersMapper;
    @Resource
    private ArticleMapper articleMapper;
    @Resource
    private ArticleSearchUtils articleSearchUtils;
    @Resource
    private JwtTokenProvider jwtTokenProvider;
    @KafkaListener(topics = "blog_topic", groupId = "blog_kafka", concurrency = "1")
    public void listen(String message) {

        try{
            StringMessageDTO stringMessageDTO = objectMapper.readValue(message, new TypeReference<StringMessageDTO>() {});
            String messageType = stringMessageDTO.getMessageType();
            if(messageType.equals(INITESDATA)) {
                InitEsDate(stringMessageDTO);
            } else if(messageType.equals(ADDARTICLEANDES)) {
                AddArticleAndEs(stringMessageDTO);
            }
        } catch (Exception e) {

            log.error("处理失败，暂停提交偏移量", e);
        }

    }

    public void InitEsDate(StringMessageDTO stringMessageDTO) {
        String articlemessage = stringMessageDTO.getArticlemessage();
        List<Article> articles = null;
        try {
            articles = objectMapper.readValue(articlemessage,new TypeReference<List<Article>>() {});
            Integer pageNum = stringMessageDTO.getMessage2();
            asyncArticleSyncService.syncBatchToEs(articles);
            log.info("第{}篇文章同步到es,大小为{}",pageNum,articles.size());
        } catch (JsonProcessingException e) {
            log.error("文章同步失败",e);
            throw new RuntimeException(e);
        }
    }
    public void AddArticleAndEs(StringMessageDTO stringMessageDTO) throws JsonProcessingException {
        String stringarticledto = stringMessageDTO.getArticlemessage();
        ArticleDTO articledto = objectMapper.readValue(stringarticledto,new TypeReference<ArticleDTO>(){});
        List<Integer> tagids = articledto.getTags();
        Integer id = UUID.randomUUID().hashCode();
        blogTagMapper.addBlogTagList(id,tagids);
        String email = articledto.getEmail();
        Users users = usersMapper.findByEmail(email);
        Article article = Article.builder()
                .id(id)
                .title(articledto.getTitle())
                .content(articledto.getContent())
                .createTime(LocalDateTime.now())
                .viewCount(0)
                .categoryId(articledto.getCategoryId())
                .status(2)
                .author(users.getNickname())
                .authorId(users.getId())
                .build();
        articleMapper.insertarticle(article);
        //将文章加入es里面
//        articleSearchUtils.saveArticle(article);
    }
}
